package io.kiki.sba.registry.server.data_store.bootstrap;

import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.common.model.slot.filter.SyncSlotAcceptAllManager;
import io.kiki.sba.registry.common.model.slot.filter.SyncSlotAcceptorManager;
import io.kiki.sba.registry.remoting.http.exchange.HttpExchanger;
import io.kiki.sba.registry.remoting.netty.NettyDataExchanger;
import io.kiki.sba.registry.server.data_store.cache.DatumStorageDelegate;
import io.kiki.sba.registry.server.data_store.change.DataChangeEventCenter;
import io.kiki.sba.registry.server.data_store.lease.ClientInterfaceLeaseManager;
import io.kiki.sba.registry.server.data_store.multi.cluster.sync.info.FetchMultiSyncService;
import io.kiki.sba.registry.server.data_store.providedata.CompressDatumService;
import io.kiki.sba.registry.server.data_store.providedata.FetchStopPushService;
import io.kiki.sba.registry.server.data_store.remoting.DataCentralControlServerManager;
import io.kiki.sba.registry.server.data_store.remoting.DataNodeExchanger;
import io.kiki.sba.registry.server.data_store.remoting.SessionNodeExchanger;
import io.kiki.sba.registry.server.data_store.remoting.dataserver.handler.SlotFollowerDiffDigestRequestHandler;
import io.kiki.sba.registry.server.data_store.remoting.dataserver.handler.SlotFollowerDiffPublisherRequestHandler;
import io.kiki.sba.registry.server.data_store.remoting.metaserver.MetaServerServiceImpl;
import io.kiki.sba.registry.server.data_store.remoting.metaserver.handler.NotifyProvideDataChangeHandler;
import io.kiki.sba.registry.server.data_store.remoting.metaserver.handler.RemoteDatumClearEventHandler;
import io.kiki.sba.registry.server.data_store.remoting.metaserver.provideData.ProvideDataProcessorManager;
import io.kiki.sba.registry.server.data_store.remoting.metaserver.provideData.processor.SessionLeaseProvideDataProcessor;
import io.kiki.sba.registry.server.data_store.remoting.sessionserver.handler.BatchPutDataHandler;
import io.kiki.sba.registry.server.data_store.remoting.sessionserver.handler.GetDataHandler;
import io.kiki.sba.registry.server.data_store.remoting.sessionserver.handler.GetDataVersionsHandler;
import io.kiki.sba.registry.server.data_store.remoting.sessionserver.handler.GetMultiDataHandler;
import io.kiki.sba.registry.server.data_store.resource.DataDigestResource;
import io.kiki.sba.registry.server.data_store.resource.DatumApiResource;
import io.kiki.sba.registry.server.data_store.resource.HealthResource;
import io.kiki.sba.registry.server.data_store.resource.SlotsStatusResource;
import io.kiki.sba.registry.server.data_store.slot.SlotAccessorDelegate;
import io.kiki.sba.registry.server.data_store.slot.SlotChangeListenerManager;
import io.kiki.sba.registry.server.data_store.slot.SlotManager;
import io.kiki.sba.registry.server.data_store.slot.SlotManagerImpl;
import io.kiki.sba.registry.server.data_store.timer.CacheCountTask;
import io.kiki.sba.registry.server.data_store.timer.CacheDigestTask;
import io.kiki.sba.registry.server.shared.config.CommonConfig;
import io.kiki.sba.registry.server.shared.meta.CentralControlLeaderExchanger;
import io.kiki.sba.registry.server.shared.providedata.FetchSystemPropertyService;
import io.kiki.sba.registry.server.shared.providedata.ProvideDataProcessor;
import io.kiki.sba.registry.server.shared.providedata.SystemPropertyProcessorManager;
import io.kiki.sba.registry.server.shared.remoting.AbstractClientHandler;
import io.kiki.sba.registry.server.shared.remoting.AbstractServerHandler;
import io.kiki.sba.registry.server.shared.remoting.SlotsChangeEventHandler;
import io.kiki.sba.registry.server.shared.resource.MetricsResource;
import io.kiki.sba.registry.server.shared.resource.RegistryOpsResource;
import io.kiki.sba.registry.server.shared.resource.SlotGenericResource;
import io.kiki.sba.registry.server.shared.resource.VersionResource;
import io.kiki.sba.registry.server.shared.slot.DiskSlotsRecorder;
import io.kiki.sba.registry.store.api.config.StoreApiConfiguration;
import io.kiki.sba.registry.store.jdbc.config.JdbcConfiguration;
import io.kiki.sba.registry.task.MetricsableThreadPoolExecutor;
import io.kiki.sba.registry.util.NamedThreadFactory;
import io.kiki.sba.registry.util.PropertySplitter;
import org.glassfish.jersey.jackson.JacksonFeature;
import org.glassfish.jersey.server.ResourceConfig;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;

import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


@Configuration
@Import({DataServerInitializer.class, StoreApiConfiguration.class, JdbcConfiguration.class,})
@EnableConfigurationProperties
public class DataServerBeanConfiguration {

    @Bean
    @ConditionalOnMissingBean
    public Bootstrap dataServerBootstrap() {
        return new Bootstrap();
    }

    @Configuration
    public static class DataServerBootstrapConfigConfiguration {

        @Bean
        public CommonConfig commonConfig() {
            return new CommonConfig();
        }

        @Bean
        @ConditionalOnMissingBean
        public ServerConfig dataServerConfig(CommonConfig commonConfig) {
            return new ServerConfig(commonConfig);
        }

        @Bean(name = "PropertySplitter")
        public PropertySplitter propertySplitter() {
            return new PropertySplitter();
        }
    }

    @Configuration
    public static class DataServerStorageConfiguration {

        @Bean
        @ConditionalOnMissingBean
        public DatumStorageDelegate datumStorageDelegate(ServerConfig serverConfig) {
            return new DatumStorageDelegate(serverConfig);
        }

        @Bean
        @ConditionalOnMissingBean
        public SlotManager slotManager() {
            return new SlotManagerImpl();
        }

        @Bean
        @ConditionalOnMissingBean
        public SlotChangeListenerManager slotChangeListenerManager() {
            return new SlotChangeListenerManager();
        }

        @Bean
        @ConditionalOnMissingBean
        public SlotAccessorDelegate slotAccessorDelegate() {
            return new SlotAccessorDelegate();
        }

        @Bean
        public SyncSlotAcceptorManager syncSlotAcceptAllManager() {
            return new SyncSlotAcceptAllManager();
        }

        @Bean
        @ConditionalOnMissingBean
        public ClientInterfaceLeaseManager sessionLeaseManager() {
            return new ClientInterfaceLeaseManager();
        }

        @Bean
        public DiskSlotsRecorder diskSlotsRecorder() {
            return new DiskSlotsRecorder();
        }
    }

    @Configuration
    public static class LogTaskConfigConfiguration {

        @Bean
        public CacheDigestTask cacheDigestTask() {
            return new CacheDigestTask();
        }

        @Bean
        public CacheCountTask cacheCountTask() {
            return new CacheCountTask();
        }
    }

    @Configuration
    public static class SessionRemotingConfiguration {

        @Bean
        public DataExchanger jerseyExchange() {
            return new HttpExchanger();
        }

        @Bean
        public DataExchanger boltExchange() {
            return new NettyDataExchanger();
        }

        @Bean
        public DataNodeExchanger dataNodeExchanger() {
            return new DataNodeExchanger();
        }

        @Bean
        public SessionNodeExchanger sessionNodeExchanger() {
            return new SessionNodeExchanger();
        }

        @Bean(name = "serverHandlers")
        public Collection<AbstractServerHandler> serverHandlers() {
            Collection<AbstractServerHandler> list = new ArrayList<>();
            list.add(getDataHandler());
            list.add(getMultiDataHandler());
            list.add(batchPutDataHandler());
            list.add(getDataVersionsHandler());
            return list;
        }

        @Bean(name = "serverSyncHandlers")
        public Collection<AbstractServerHandler> serverSyncHandlers() {
            Collection<AbstractServerHandler> list = new ArrayList<>();
            list.add(slotFollowerDiffDataInfoIdRequestHandler());
            list.add(slotFollowerDiffPublisherRequestHandler());
            return list;
        }

        @Bean(name = "metaClientHandlers")
        public Collection<AbstractClientHandler> metaClientHandlers() {
            Collection<AbstractClientHandler> list = new ArrayList<>();
            list.add(notifyProvideDataChangeHandler());
            list.add(SlotsChangeEventHandler());
            list.add(remoteDatumClearEventHandler());
            return list;
        }

        @Bean
        public AbstractServerHandler getDataHandler() {
            return new GetDataHandler();
        }

        @Bean
        public AbstractServerHandler getMultiDataHandler() {
            return new GetMultiDataHandler();
        }

        @Bean
        public AbstractServerHandler slotFollowerDiffDataInfoIdRequestHandler() {
            return new SlotFollowerDiffDigestRequestHandler();
        }

        @Bean
        public AbstractServerHandler slotFollowerDiffPublisherRequestHandler() {
            return new SlotFollowerDiffPublisherRequestHandler();
        }

        @Bean
        public AbstractServerHandler getDataVersionsHandler() {
            return new GetDataVersionsHandler();
        }

        @Bean
        public AbstractServerHandler batchPutDataHandler() {
            return new BatchPutDataHandler();
        }

        @Bean
        public NotifyProvideDataChangeHandler notifyProvideDataChangeHandler() {
            return new NotifyProvideDataChangeHandler();
        }

        @Bean
        public SlotsChangeEventHandler SlotsChangeEventHandler() {
            return new SlotsChangeEventHandler();
        }

        @Bean
        public RemoteDatumClearEventHandler remoteDatumClearEventHandler() {
            return new RemoteDatumClearEventHandler();
        }
    }

    @Configuration
    public static class DataServerEventBeanConfiguration {

        @Bean
        public DataChangeEventCenter dataChangeEventCenter() {
            return new DataChangeEventCenter();
        }
    }

    @Configuration
    public static class DataServerRemotingBeanConfiguration {

        @Bean
        public CentralControlLeaderExchanger metaLeaderExchanger() {
            return new DataCentralControlServerManager();
        }

        @Bean
        public MetaServerServiceImpl metaServerService() {
            return new MetaServerServiceImpl();
        }

        @Bean
        public FetchMultiSyncService fetchMultiSyncService() {
            return new FetchMultiSyncService();
        }
    }

    @Configuration
    public static class ResourceConfiguration {

        @Bean
        public ResourceConfig jerseyResourceConfig() {
            ResourceConfig resourceConfig = new ResourceConfig();
            resourceConfig.register(JacksonFeature.class);
            return resourceConfig;
        }

        @Bean
        public HealthResource healthResource() {
            return new HealthResource();
        }

        @Bean
        public DatumApiResource datumApiResource() {
            return new DatumApiResource();
        }

        @Bean
        @ConditionalOnMissingBean
        public DataDigestResource dataDigestResource() {
            return new DataDigestResource();
        }

        @Bean
        public SlotGenericResource slotGenericResource() {
            return new SlotGenericResource();
        }

        @Bean
        public MetricsResource metricsResource() {
            return new MetricsResource();
        }

        @Bean
        public RegistryOpsResource opsResource() {
            return new RegistryOpsResource();
        }

        @Bean
        @ConditionalOnMissingBean
        public VersionResource versionResource() {
            return new VersionResource();
        }

        @Bean
        public SlotsStatusResource SlotsResource() {
            return new SlotsStatusResource();
        }
    }

    @Configuration
    public static class ExecutorConfiguration {

        @Bean(name = "publishProcessorExecutor")
        public ThreadPoolExecutor publishProcessorExecutor(ServerConfig serverConfig) {
            return new MetricsableThreadPoolExecutor("PublishProcessorExecutor", serverConfig.getPublishExecutorMinPoolSize(), serverConfig.getPublishExecutorMaxPoolSize(), 300, TimeUnit.SECONDS, new ArrayBlockingQueue<>(serverConfig.getPublishExecutorQueueSize()), new NamedThreadFactory("PutExecutor", true));
        }

        @Bean(name = "getDataProcessorExecutor")
        public ThreadPoolExecutor getDataProcessorExecutor(ServerConfig serverConfig) {
            return new MetricsableThreadPoolExecutor("GetDataProcessorExecutor", serverConfig.getGetDataExecutorMinPoolSize(), serverConfig.getGetDataExecutorMaxPoolSize(), serverConfig.getGetDataExecutorKeepAliveTime(), TimeUnit.SECONDS, new ArrayBlockingQueue<>(serverConfig.getGetDataExecutorQueueSize()), new NamedThreadFactory("GetExecutor", true));
        }

        @Bean(name = "slotSyncRequestProcessorExecutor")
        public ThreadPoolExecutor slotSyncRequestProcessorExecutor(ServerConfig serverConfig) {
            return new MetricsableThreadPoolExecutor("SlotSyncRequestProcessorExecutor", serverConfig.getSlotSyncRequestExecutorMinPoolSize(), serverConfig.getSlotSyncRequestExecutorMaxPoolSize(), 300, TimeUnit.SECONDS, new ArrayBlockingQueue<>(serverConfig.getSlotSyncRequestExecutorQueueSize()), new NamedThreadFactory("SyncExecutor", true));
        }

        @Bean(name = "metaNodeExecutor")
        public ThreadPoolExecutor metaNodeExecutor(ServerConfig serverConfig) {
            return new MetricsableThreadPoolExecutor("metaNodeInDataExecutor", serverConfig.getMetaNodeExecutorPoolSize(), serverConfig.getMetaNodeExecutorPoolSize(), 300, TimeUnit.SECONDS, new ArrayBlockingQueue<>(serverConfig.getMetaNodeExecutorQueueSize()), new NamedThreadFactory("metaNodeInDataExecutor", true));
        }
    }

    @Configuration
    public static class DataProvideDataConfiguration {

        @Bean
        public ProvideDataProcessor provideDataProcessorManager() {
            return new ProvideDataProcessorManager();
        }

        @Bean
        public ProvideDataProcessor sessionLeaseProvideDataProcessor(ProvideDataProcessor provideDataProcessorManager) {
            ProvideDataProcessor sessionLeaseProvideDataProcessor = new SessionLeaseProvideDataProcessor();
            ((ProvideDataProcessorManager) provideDataProcessorManager).addProvideDataProcessor(sessionLeaseProvideDataProcessor);
            return sessionLeaseProvideDataProcessor;
        }

        @Bean
        public SystemPropertyProcessorManager systemPropertyProcessorManager() {
            return new SystemPropertyProcessorManager();
        }

        @Bean
        public FetchSystemPropertyService compressDatumService(SystemPropertyProcessorManager systemPropertyProcessorManager) {
            CompressDatumService compressDatumService = new CompressDatumService();
            systemPropertyProcessorManager.addSystemDataProcessor(compressDatumService);
            return compressDatumService;
        }

        @Bean
        public FetchSystemPropertyService fetchStopPushService(SystemPropertyProcessorManager systemPropertyProcessorManager) {
            FetchStopPushService fetchStopPushService = new FetchStopPushService();
            systemPropertyProcessorManager.addSystemDataPersistenceProcessor(fetchStopPushService);
            return fetchStopPushService;
        }
    }
}
